package david.java.flink_sql.streamApiIntegration;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Description:  to DataStream
 * @link https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#converting-between-datastream-and-table
 * @Author: ZhaoDawei
 * @Date: Create in 10:19 上午 2022/3/22
 */
public class S1_StreamTableConversion_1 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> dataStream = env.fromElements("Alice, Bob, John");

        Table inputTable = tableEnv.fromDataStream(dataStream);

        tableEnv.createTemporaryView("InputTable", inputTable);
        // Table result = tableEnv.sqlQuery("select f0 from InputTable");
        // upper 的作用是 小写转大写
        // +----+--------------------------------+
        // | op |                         EXPR$0 |
        // +----+--------------------------------+
        // | +I |               ALICE, BOB, JOHN |
        // +----+--------------------------------+
        Table result = tableEnv.sqlQuery("select UPPER(f0) from InputTable");

        result.printSchema();
        TableResult execute = result.execute();
        execute.print();

        // 加上这句会报错 问题就在这里，由于executeSql已经是异步提交了作业，生成Transformation后会把缓存的Operation清除，
        // 见TableEnvironmentImpl#translateAndClearBuffer,执行execute也会走那一段逻辑，报了上面异常，但是这个异常不影响程序执行。
        // env.execute();


        DataStream<Row> stream = tableEnv.toDataStream(result);
        // 5> +I[ALICE, BOB, JOHN]
        stream.print();
        env.execute();   // 这里就要加 此方法了,否则不会执行, stream 不会打印;





    }
}
